package com.allwees.core.mq.rabbitmq.batch;

import com.allwees.core.common.util.json.JsonUtil;
import com.allwees.core.common.util.reflect.ReflectionUtils;
import org.springframework.amqp.core.Message;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 批量消息消费者
 *
 * @author yangylsky
 * @since 2019-11-25
 *
 * @param <M> - Mq消息解析出来的类型
 * @param <S> - 实际消费处理的目标类型
 */
public abstract class AbstractConsumer<M, S> {
	Class<M> mClass;

	public AbstractConsumer() {
		super();
		mClass = ReflectionUtils.getGenricParameterType(this);
	}

	/**
	 * 解析Mq消息
	 *
	 * @param msg
	 * @return
	 * @throws UnsupportedEncodingException
	 */
	M parseMessage(Message msg) throws UnsupportedEncodingException {
		String json = new String(msg.getBody(), StandardCharsets.UTF_8.name());
		M m = JsonUtil.fromJson(json, mClass);
		return m;
	}

	/**
	 * 转换待处理消息到目标类型
	 *
	 * @param m
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public S transform(M m) {
		return (S) m;
	}

	/**
	 * 执行批量消费动作
	 *
	 * @param list
	 */
	public abstract void consume(List<S> list);

	/**
	 * 消费完成的后续处理
	 *
	 * @param dealNum - 实际消费的数量
	 * @param processMillis - 处理耗时
	 */
	public abstract void afterConsume(int dealNum, long processMillis);
}
